package com.gzsxy.esjy.message.channel;

import com.alibaba.fastjson.JSON;
import com.gzsxy.esjy.redis.extend.RedisLock;
import com.gzsxy.esjy.redis.extend.impl.DefaultDistributeLock;
import com.gzsxy.esjy.redis.extend.impl.Lock;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import java.util.*;
import java.util.Map.Entry;

/**
 * 默认通道管理
 * @author zhangxiaolong
 *
 */
@Component
@Slf4j
public class DefaultChannelManager implements ChannelManager {

	private final static Map<String, ChannelGroup> GROUPS = new HashMap<String, ChannelGroup>();

	private final static String GROUP_STR = "com.gosuncn.ivmp.messagecenter.channel" ;

	@Autowired(required=false)
	private RedisLock lock ;

	@PostConstruct
	private void init(){
		if(lock == null){
			lock = new DefaultDistributeLock(24) ;
		}
	}

	/**
	 * 绑定关闭事件，将通道从订阅主题中移除
	 */
	private final static ChannelFutureListener remover = new ChannelFutureListener() {
		@Override
		public void operationComplete(ChannelFuture future) {
			Channel channel = future.channel();
			Set<Entry<String, ChannelGroup>> entrySet = GROUPS.entrySet();
			entrySet.forEach(entity -> entity.getValue().remove(channel));
		}
	};

	@Override
	public ChannelGroup getChannel(String topic) {
		return GROUPS.get(topic);
	}

	@Override
	public void addChannel(String topic, Channel channel) {
		if(StringUtils.hasText(topic)) {
			lock.excute(new Lock(topic,GROUP_STR),(k)->{
				ChannelGroup channelGroup = GROUPS.get(topic);
				//如果该主题下没有通道组
				if (channelGroup == null) {
					//创建一个默认的通道组
					channelGroup = new DefaultChannelGroup(topic, GlobalEventExecutor.INSTANCE);
					GROUPS.put(topic, channelGroup);
				}
				//将该通道添加入通道组
				channelGroup.add(channel);
			}
			);
		}
	}
	
	
	public void console() {
		Set<String> keySet = GROUPS.keySet();
		log.info("当前订阅情况:");
		List<Object> args = new ArrayList<>() ;
		StringBuilder builder = new StringBuilder() ;
		List<String> keys = new ArrayList<String>() ;
		for (String key : keySet) {
			ChannelGroup channelGroup = GROUPS.get(key);
			if(channelGroup != null) {
				int size = channelGroup.size();
				builder.append("主题：{}  订阅客户数：{} {}\n");
				args.add(key);
				args.add(size);
				Iterator<Channel> iterator = channelGroup.iterator();
				List<String> list = new ArrayList<>() ;
				if(size > 0) {
					while (iterator.hasNext()) {
						list.add(iterator.next().attr(ChannelManager.IP).get());
					}
				}
				args.add(JSON.toJSONString(list));
				if(size == 0) {
					keys.add(key);
				}
			}
		}
		keys.forEach(key->lock.excute(new Lock(key,GROUP_STR),(k)->GROUPS.remove(key)));
		Object[] objs = new Object[args.size()];
		args.toArray(objs);
		log.info(builder.toString(),objs);
	}


	@Override
	public void remove(Channel channel) {
		channel.close() ;
	}

	@Override
	public void remove(String topic, Channel channel) {
		ChannelGroup channelGroup = GROUPS.get(topic);
		if(channelGroup != null) {
			channelGroup.remove(channel) ;
		}
	}

	@Override
	public void closeEvent(Channel channel) {
		channel.closeFuture().addListener(remover) ;
	}

}
